package com.mioto.ld.communication.mqtt;

import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.io.UnsupportedEncodingException;
import java.util.Arrays;
import java.util.stream.Collectors;

/**
 * @author mioto-qinxj
 * @date 2024/5/6
 * @desc
 */
@Slf4j
public class MqttClientConnect {
    private MqttClient mqttClient;
    /**
     * 主题前缀 /username/password
     */
    private String topicPrefix;
    /**
     * 客户端id
     */
    private Integer mqttClientId;

    /**
     * 订阅地址
     */
    private String subTopic;

    public void setMqttClient(String host, String clientId, String userName, String passWord, MqttCallback mqttCallback) throws MqttException {
        // 连接设置
        MqttConnectOptions options = mqttConnectOptions(userName, passWord);
        // 创建 MQTT 客户端对象
        mqttClient = new MqttClient(host, clientId, new MemoryPersistence());
        if (mqttCallback == null) {
            mqttClient.setCallback(new MqttClientCallback(mqttClientId));
        } else {
            mqttClient.setCallback(mqttCallback);
        }
        mqttClient.connect(options);
    }
    /**
     * MQTT连接参数设置
     */
    private MqttConnectOptions mqttConnectOptions(String userName, String passWord) throws MqttException {
        // 连接设置
        MqttConnectOptions options = new MqttConnectOptions();
        // 设置连接用户名
        options.setUserName(userName);
        // 设置连接密码
        options.setPassword(passWord.toCharArray());
        // 设置超时时间，单位为秒
        options.setConnectionTimeout(60);
        // 设置心跳时间 单位为秒，表示服务器每 60 秒向客户端发送心跳判断客户端是否在线
        options.setKeepAliveInterval(60);
        // 设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息
        //options.setWill("willTopic", null, 0, false);
        // 断线自动重连
        options.setAutomaticReconnect(true);
        // 是否清空 session，设置为 false 表示服务器会保留客户端的连接记录，客户端重连之后能获取到服务器在客户端断开连接期间推送的消息
        // 设置为 true 表示每次连接到服务端都是以新的身份
        options.setCleanSession(true);
        return options;
    }


    /**
     * 向某个主题发布消息
     * @param sn 设备序列号
     * @param msg 发布的消息
     */
    public void pub(String sn, String msg) {
        MqttMessage mqttMessage = new MqttMessage();
        try {
            mqttMessage.setPayload(msg.getBytes("UTF-8"));
        } catch (UnsupportedEncodingException e) {
            log.error(e.getMessage(),e);
        }
        String pubTopic = StrUtil.format("{}/{}/IOT/get",topicPrefix,sn);
        MqttTopic mqttTopic = mqttClient.getTopic(pubTopic);
        try {
            MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
            token.waitForCompletion();
            log.info("发布主题 - {} 消息成功 - {}",pubTopic,msg);
        } catch (MqttException e) {
            log.error("消息发布失败 - {}",msg,e.getMessage(),e);
        }
    }

    /**
     * 订阅主题
     * @param topic 主题
     * @throws MqttException
     */
    public void sub(String topic) throws MqttException {
        this.subTopic = topic;
        topicPrefix = Arrays.stream(StrUtil.split(topic, StrUtil.SLASH)).limit(3).collect(Collectors.joining( StrUtil.SLASH));
        mqttClient.subscribe(topic);
    }


    /**
     * 关闭MQTT连接
     */
    public void close(){
        try {
            mqttClient.close();
            mqttClient.disconnect();
        }catch (MqttException e){
            log.error("mqtt连接关闭失败 - {}",e.getMessage(),e);
        }

    }

    public void setMqttClientId(Integer mqttClientId) {
        this.mqttClientId = mqttClientId;
    }

    public Integer getMqttClientId() {
        return mqttClientId;
    }


    public String getSubTopic(){
        return this.subTopic;
    }
}
